package com.dyna.cp.kafka.service.impl;

import com.alibaba.fastjson.JSON;
import com.dyna.constants.TopicConstant;
import com.dyna.cp.kafka.service.KafkaService;
import com.dyna.demo.device.domain.entity.DeviceInfo;
import com.dyna.domain.resp.PageResult;
import com.dyna.utils.EsUtil;
import com.dyna.utils.KafkaUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@Service
@RefreshScope
public class KafkaServiceImpl implements KafkaService {

    @Value("${dyna.username}")
    private String dynaUsername;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void testKafka(String msg) {
        log.info("dynaUsername:::{}", dynaUsername);
        KafkaUtil.send(TopicConstant.TEST_TOPIC, msg);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void kafkaRollback(String msg) {
        KafkaUtil.send(TopicConstant.TEST_TOPIC, msg);
    }

    /**
     * 不管使用kafka事务配置还是不使用，kafka都能顺序消费掉消息
     */
    @Override
    public void orderConsumerMsg() {
        String name = Thread.currentThread().getName();
        List<String> list = new ArrayList<>();
        list.add("1");
        list.add("2");
        list.add("3");
        list.add("4");
        list.add("5");
        for (String msg : list) {
            KafkaUtil.send(TopicConstant.TEST_TOPIC, name + "---" + msg);
        }
    }

    @Override
    public void sendDesignatePartition(String topic, Integer partition, String key, String msg) {
        KafkaUtil.send(topic, partition, key, msg);
    }

    @Override
    public void batchProductMsg(List<String> msgList) {
        PageResult<DeviceInfo> pageResult = EsUtil.queryDataAll("dc_device_info", DeviceInfo.class);
        System.out.println("查询数量：" + pageResult.getTotal());
        System.out.println("查询结果：" + pageResult.getList());

        // 发送给kafka
        long startTime = System.currentTimeMillis();
        for (DeviceInfo map : pageResult.getList()) {
            KafkaUtil.send(TopicConstant.TEST_TOPIC_2, 1, "test-topic-2-key", JSON.toJSONString(map));
        }
        System.out.println("发送耗时：" + (System.currentTimeMillis() - startTime));
        System.out.println("发送数量：" + pageResult.getList().size());
    }

}
